package com.xxmw.transaction.service.app.component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xxmw.transaction.common.utils.HuoBiGZipUtils;
import com.xxmw.transaction.enumeration.BourseEnum;
import com.xxmw.transaction.model.app.BoursePrice;
import com.xxmw.transaction.model.app.HuoBiTxPair;
import com.xxmw.transaction.model.app.Symbol;
import com.xxmw.transaction.service.app.HuoBiTxPairService;
import com.xxmw.transaction.service.app.SymbolService;
import com.xxmw.transaction.service.support.AbstractWebSocketAdapter;
import com.xxmw.transaction.service.support.SymbolSubWebSocket;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.drafts.Draft_6455;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public final class HuoBiTickerApplication extends AbstractWebSocketAdapter implements SymbolSubWebSocket {

	private static final BlockingQueue<HuoBiTxPair> BQ = new LinkedBlockingQueue<>(500);

	private static final String HB_URL = "wss://api.hadax.com/ws";
//	private static final String HB_URL = "wss://api.huobi.pro/ws";

	private static final int FREQMS_VALUE = 5000;

	/**
	 * 执行心跳的间隔时间和在此基础商上的超时时间将由实现的子类所设定(或者以配置的方式来是定)
	 * 假如：此时间有父类适配器所设定那么所有共同子类将继承，如需改变其中一个子类的时间，将会造成
	 * 所有子类间隔时间更改
	 */
	private long period = 3000;

	@Autowired
	private HuoBiTxPairService huoBiTxPairService;

	@Autowired
	private SymbolService symbolService;

	@Override
	public void afterPropertiesSet() throws Exception {
		super.afterPropertiesSet(new HuoBiWebSocketClient(new URI(HB_URL), new Draft_6455(), null, 5000));
		scheduled.scheduleWithFixedDelay(() -> heartBeat(), period, period, TimeUnit.MILLISECONDS);
		scheduled.scheduleWithFixedDelay(() -> saveAllTxPair(), 0, 5 , TimeUnit.MINUTES);
	}

	@Override
	public void destroy() throws Exception {
		this.client.close();
		scheduled.shutdown();
	}

	private void saveAllTxPair() {
		List<HuoBiTxPair> list = new ArrayList<>();
		while(BQ.size() > 0) {
			HuoBiTxPair biTxPair = BQ.poll();
			biTxPair.setCreated(new Date());
			list.add(biTxPair);
		}
		huoBiTxPairService.saveBatch(list);
	}

	@Override
	public void subscribe(Symbol symbol) {
		//火币订阅事件
		JSONObject object = new JSONObject();
		object.put("sub", "market."+ BourseEnum.HUOBI.getSymbolFormat().symbolFormat(symbol) +".bbo");
		object.put("id", sequence.nextId());
		//object.put(FREQMS_KEY, FREQMS_VALUE);
		this.client.send(object.toJSONString());
		//订阅交易对市场详情，用于获取最新价
		JSONObject o = new JSONObject();
		o.put("sub", "market." + BourseEnum.HUOBI.getSymbolFormat().symbolFormat(symbol) + ".detail");
		o.put("id", sequence.nextId());
		this.client.send(o.toJSONString());
	}

	/*public void subscribe(String symbol) {
		//火币订阅事件
		JSONObject object = new JSONObject();
		object.put("sub", "market."+ symbol +".bbo");
		object.put("id", sequence.nextId());
		//object.put(FREQMS_KEY, FREQMS_VALUE);
		this.client.send(object.toJSONString());
	}*/

	@Override
	public void unSubscribe(Symbol symbol) {
		//火币取消订阅；
		JSONObject object = new JSONObject();
		object.put("unsub", "market."+BourseEnum.HUOBI.getSymbolFormat().symbolFormat(symbol)+".bbo");
		object.put("id", sequence.nextId());
		//object.put(FREQMS_KEY, FREQMS_VALUE);
		this.client.send(object.toJSONString());

		cleanSymbolPrice(symbol.getName());
	}

	@Override
	public void subscribeList(Set<Symbol> subSet) {
		for (Symbol symbol: subSet) {
			subscribe(symbol);
		}
	}

	@Override
	public BoursePrice getSymbolPrice (String symbol) {
		return symbol_map.get(symbol);
	}

	@Override
	public void cleanSymbolPrice(String symbol) {
		symbol_map.remove(symbol);
	}

	private final class HuoBiWebSocketClient extends WebSocketClient {

		public HuoBiWebSocketClient(URI serverUri, Draft protocolDraft, Map<String, String> httpHeaders,
				int connectTimeout) {
			super(serverUri, protocolDraft, httpHeaders, connectTimeout);
		}

		@Override
		public void onOpen(ServerHandshake handshakedata) {
			symbolService.cacheSymbolDayOpenPrice();
			log.info("火币WebSocket接口: {} 链接成功.", this.getURI());
			Set<Symbol> subSet = symbolService.listByBourse(BourseEnum.HUOBI.getValue());
			if (subSet != null && !subSet.isEmpty()) {
				//Set<String> topicSet = subSet.stream().map(item -> BourseEnum.HUOBI.getSymbolFormat().symbolFormat(item)).collect(Collectors.toSet());
				log.info("火币订阅开始{}",subSet);
				subscribeList(subSet);
			}
		}

		@Override
		public void onMessage(ByteBuffer bytes) {
			byte[] array = bytes.array();
			array = HuoBiGZipUtils.decompress(array);
			String str = new String(array, StandardCharsets.UTF_8);
			if(str.contains(PING)) {
				this.send(str.replace(PING, PONG));
			}else if(str.contains("tick")){
				JSONObject object = JSON.parseObject(str);

				//订阅主题
				String ch = object.getString("ch");

				JSONObject data = object.getJSONObject("tick");

				if (data.containsKey("bid")) {
					String symbol = data.getString("symbol");
					data.put("updateTime",new Date());
//					log.info("火币{}获取买一价格：{} 卖一价格：{}",symbol, data.getString("bid"), data.getString("ask"));
					BoursePrice price = BoursePrice.builder().buyPrice(data.getBigDecimal("bid")).buyNum(data.getBigDecimal("bidSize")).sellPrice(data.getBigDecimal("ask")).sellNum(data.getBigDecimal("askSize")).data(data.toJSONString()).updateTime(new Date()).build();
					//  log.info("火币获取到【***{}***】的实时价格为：{}", symbol, price.toString());
					symbol_map.put(symbol, price);
				}
				if (data.containsKey("close")) {

					final String[] split = ch.split("\\.");
					symbolService.pushSymbolInfo(split[1], data.getString("close"));
				}
			}else {
				log.error("火币WebSocket数据: {} 未做处理", str);
			}
		}

		@Override
		public void onMessage(String message) {
			log.info("接收到火币的信息：{}",message);
		}

		@Override
		public void onClose(int code, String reason, boolean remote) {
			log.info("火币Websocket接口已关闭, 原因: {}, 状态: {}", reason, code);
		}

		@Override
		public void onError(Exception e) {
			log.error("火币WebSocket接口，发生异常", e);
		}


	}

}
